package com.hujf.flink.window.TimeWindow;

import com.hujf.flink.ch05.MySource;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * @Author hujf
 * @Date 2022/5/23 16:13
 * @Description TODO
 */
public class SessionWindowDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> dataStreamSource = environment.addSource(new MySource());

        SingleOutputStreamOperator<Tuple2<String, Integer>> sum = dataStreamSource.flatMap(new Splitter())
                .keyBy(0)
                .window(ProcessingTimeSessionWindows.withGap(Time.seconds(20)))
                .sum(1);

        sum.print();
        environment.execute();

    }

    private static class Splitter implements FlatMapFunction<String, Tuple2<String,Integer>> {
        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
            for(String w:value.split(" " )){
                out.collect(new Tuple2<>(w,1));
            }
        }
    }
}
